/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hylanda.processors;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.nifi.annotation.behavior.CustomDataPreferred;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.ShowState;
import org.apache.nifi.annotation.behavior.StateKey;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.InputStreamCallback;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.LogAppendInfoUtil;

import com.google.common.collect.Maps;
import com.hylanda.tools.HttpTools;

/***
 * 该能力的主要功能是集成一个自定义的html+javascript页面，实现数据可视化
 * 
 * @author song
 *
 */

@Tags({ "自定义可视化-JS" })
@CapabilityDescription("通过自定义的html+javascript页面，实现数据可视化")
@SeeAlso({})
@ReadsAttributes({ @ReadsAttribute(attribute = "", description = "") })
@WritesAttributes({ @WritesAttribute(attribute = "", description = "") })
@CustomDataPreferred
@ShowState(keys = { @StateKey(key = "aa") })
public class HLCustomVisualProcessor extends AbstractProcessor {

//	public static final PropertyDescriptor PROPERTY_USAGE = new PropertyDescriptor.Builder().name("usage")
//			.displayName("启动说明").description("请连接统计能力，获取统计结果，然后启动能力，查看数据页").required(false)
//			.defaultValue("请连接统计能力，获取统计结果，然后启动能力，查看数据页").addValidator(Validator.VALID).build();
	
	public static final PropertyDescriptor bar_p1 = new PropertyDescriptor.Builder().name("bar_p1")
			.displayName("统计字段(必填)").description("统计字段(必填)").required(true)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
	public static final PropertyDescriptor bar_p2 = new PropertyDescriptor.Builder().name("bar_p2")
			.displayName("图表标题").description("图表标题").required(true)
			.defaultValue("").addValidator(Validator.VALID).build();
	public static final PropertyDescriptor bar_p3 = new PropertyDescriptor.Builder().name("bar_p3")
			.displayName("图表副标题").description("图表副标题").required(true)
			.defaultValue("").addValidator(Validator.VALID).build();
	public static final PropertyDescriptor bar_p4 = new PropertyDescriptor.Builder().name("bar_p4")
			.displayName("计数项名称").description("计数项名称").required(true)
			.defaultValue("").addValidator(Validator.VALID).build();
	public static final PropertyDescriptor bar_p5 = new PropertyDescriptor.Builder().name("bar_p5")
			.displayName("图例名称").description("图例名称").required(true)
			.defaultValue("").addValidator(Validator.VALID).build();
	
	public static final PropertyDescriptor pie_p1 = new PropertyDescriptor.Builder().name("pie_p1")
			.displayName("统计字段(必填)").description("统计字段(必填)").required(true)
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
	public static final PropertyDescriptor pie_p2 = new PropertyDescriptor.Builder().name("pie_p2")
			.displayName("图表标题").description("图表标题").required(true)
			.defaultValue("").addValidator(Validator.VALID).build();
	
	public static final Relationship RELATIONSHIP_OK = new Relationship.Builder().name("成功").description("成功")
			.autoTerminateDefault(true).build();
	public static final Relationship RELATIONSHIP_ERR = new Relationship.Builder().name("失败").description("失败")
			.autoTerminateDefault(true).build();

	private List<PropertyDescriptor> descriptors;
	private Set<Relationship> relationships;
	private String v_type = "DEFAULT_HTML";
	// private String m_uuid = null;
	// private static final ReentrantLock startWorkLock = new ReentrantLock();

	// 参数列表
	
	@Override
	protected void init(final ProcessorInitializationContext context) {
		final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
		if(context.getExtendProperties() != null){
			// CustomVisualWordCloud 词云图
			if("e00c220ddd2c4f3426aa8ede9d006492".equals(context.getExtendProperties().get("hidden_processor_id"))){
				v_type = "WORD_CLOUD";
			}
			// CustomVisualBar
			else if("a0f3f94cd40651644778cf95a2cfeda0".equals(context.getExtendProperties().get("hidden_processor_id"))){
				v_type = "BAR";
				descriptors.add(bar_p1);
				descriptors.add(bar_p2);
				descriptors.add(bar_p3);
				descriptors.add(bar_p4);
				descriptors.add(bar_p5);
			}
			// CustomVisualPie
			else if("d954c735c9c5a7daeb70e7dcd584d3c8".equals(context.getExtendProperties().get("hidden_processor_id"))){
				v_type = "PIE";
				descriptors.add(pie_p1);
				descriptors.add(pie_p2);
			}
		}
		this.descriptors = Collections.unmodifiableList(descriptors);

		final Set<Relationship> relationships = new HashSet<Relationship>();
		// relationships.add(RELATIONSHIP_OK);
		// relationships.add(RELATIONSHIP_ERR);
		this.relationships = Collections.unmodifiableSet(relationships);
		
	}

	@Override
	public Set<Relationship> getRelationships() {
		return this.relationships;
	}

	@Override
	public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
		return descriptors;
	}

	// 任务启动
	@OnScheduled
	public void onScheduled(final ProcessContext context) throws Exception{
		
		//if (startWorkLock.tryLock()) {
			try {

				String htmlJsSrc = null;
				String processorId = this.getIdentifier();
				boolean bNeedNewUUID = false;
				String uuid = context.getExtendProperty("cvis_uuid");
				if (uuid == null || uuid.isEmpty()) {// 没有uuid，则为全新
														// 能力使用，则设置参数，用默认值
					uuid = UUID.randomUUID().toString().toLowerCase();
					context.putExtendProperty("cvis_uuid", uuid);
					UuidToProcessorIDMap.getInstance().setCache(uuid, processorId);
				} else {// uuid 存在，
						// 需要检查是否为复制能力（或模板直接创建的），用cache的pid和当前pid是否相等判断，
					String cachePid = UuidToProcessorIDMap.getInstance().getCache(uuid);
					if (cachePid != null && !cachePid.equals(processorId)) {
						// 有cache且cache的pid不相等，需要重新产生uuid
						bNeedNewUUID = true;
						getLogger().info("custom visual bNeedNewUUID = True， cachePid="+cachePid+", curPid="+processorId);
					}else if(cachePid == null){
						// 无cache，则设置cache
						UuidToProcessorIDMap.getInstance().setCache(uuid, processorId);
					}
					getLogger().info("getHtmlFromMysql start ...");
					htmlJsSrc = getHtmlFromMysql(uuid);
					getLogger().info("getHtmlFromMysql end! html=" + htmlJsSrc);
				}

				getLogger().info("get pid,uid,stid....");

				String project_id = LogAppendInfoUtil.projectId();
				String user_id = LogAppendInfoUtil.userId();
				String studio_id = LogAppendInfoUtil.studioId();

				getLogger().info("get pid,uid,stid  end!");
				// 从数据库中读取缓存，不存在则赋予默认值。 数据库的 创建、更新、删除
				// 操作均在processor中处理，与UI模块的交互，用http请求处理，post请求，addcache，getcache
				//
				if (bNeedNewUUID) {// 更新uuid
					getLogger().info("gen uuid new ....");
					uuid = UUID.randomUUID().toString().toLowerCase();
					getLogger().info("context.putExtendProperty ....");
					context.putExtendProperty("cvis_uuid", uuid);
					
					String uuidTmp = context.getExtendProperty("cvis_uuid");
					if(uuidTmp == null){

						getLogger().error("context.getExtendProperty(\"cvis_uuid\") failed! uuid="+uuid);
					}else{
						getLogger().info("context.getExtendProperty(\"cvis_uuid\") OK! uuid="+uuid+", uuidTmp="+uuidTmp);
					}
					
					getLogger().info("UuidToProcessorIDMap.getInstance().setCache ....");
					UuidToProcessorIDMap.getInstance().setCache(uuid, processorId);
					getLogger().info("gen uuid end!");
				}
				// 获取本机IP
				getLogger().info("getWebIP start ...");
				String strWebIP = getWebIP();
				getLogger().info("getWebIP end,ip=" + strWebIP);
				String getDataUrl = "http://" + strWebIP + ":8080/nifi-hlcustomerUI-ui-1.0-SNAPSHOT/customvisualui/getcurdata?pid="
						+ processorId;

				if (htmlJsSrc == null || bNeedNewUUID) {// 新uuid，则需要插入数据库记录
					if (htmlJsSrc == null) {
						getLogger().info("getHtmlFromMysql 2 start ...");
						
						htmlJsSrc = getHtmlFromMysql(v_type);
						if(v_type.equals("BAR")){
							for(int i=1; i<=5; i++){
								htmlJsSrc = htmlJsSrc.replace("hl_params_" + i, context.getProperty("bar_p" + i).getValue());
							}
						}
						else if(v_type.equals("PIE")){
							for(int i=1; i<=2; i++){
								htmlJsSrc = htmlJsSrc.replace("hl_params_" + i, context.getProperty("pie_p" + i).getValue());
							}
						}
						
						getLogger().info("getHtmlFromMysql 2 end!");
						if (htmlJsSrc != null) {							
							htmlJsSrc = htmlJsSrc.replaceAll("_GET_DATA_URL_", getDataUrl);
						}
					}
					// 存入数据库，需要替换dataurl
					getLogger().info("addHtmlToMysql start ...");
					boolean bAdd = addHtmlToMysql(uuid, project_id, user_id, studio_id,htmlJsSrc);
					getLogger().info("addHtmlToMysql end");
					if (!bAdd) {
						getLogger().error("custom visual addHtmlToMysql failed, UUID=" + uuid);
					}
				}else{
					String reg = "[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\:8080/.*\"";
					Pattern pattern = Pattern.compile(reg);
					Matcher matcher = pattern.matcher(htmlJsSrc);
					boolean bReped = false;
					while (matcher.find()) {
						String matchWord = matcher.group(0);
						if(!matchWord.equals(getDataUrl+"\"")){
							htmlJsSrc = htmlJsSrc.replace(matchWord, getDataUrl.substring(7)+"\"");
							bReped = true;
						}
					}
					if(bReped){
						updateHtmlToMysql(uuid, htmlJsSrc);
					}
				}
				/*// 设置UI的内存cache
				getLogger().info("setUICache start...");
				boolean bSetUICache = setUICache(uuid, htmlJsSrc);
				getLogger().info("setUICache end");
				if (!bSetUICache) {
					getLogger().error("custom visual setUICache failed, UUID=" + uuid);
				}*/
				
				try {
					String dataUrl = "/nifi-hlcustomerUI-ui-1.0-SNAPSHOT/customvisualui/customvisual?uuid=" + uuid + "&pid="
							+ processorId;
					final String finUrl = dataUrl;
					getLogger().info("custom visual url: " + finUrl);
					context.getStateManager().setState(new HashMap<String, String>() {
						{
							put("data_url", finUrl);
							put("cur_data", "{\"word\":\"统计词\",\"count\":100}");
						}
					}, Scope.LOCAL);
				} catch (IOException e) {
					e.printStackTrace();
				}
			} catch (Exception e) {
				// Util.status(logger, identifier, "myConfig.ini",
				// "无法读取数据库配置文件。", identifier);
				throw e;
			} finally {
				//startWorkLock.unlock();
			}
		//}else{
		//	throw new Exception("not have lock!");
		//}
	}

	@OnStopped
	public void onStopped(final ProcessContext context) {
		// 保存到数据库
		String uuid = context.getExtendProperty("cvis_uuid");
		if (uuid != null && !uuid.isEmpty()) {
			// 更新html到数据库
			String htmlSrc = getUICache(uuid);
			if (htmlSrc != null) {

				boolean bUpdate = updateHtmlToMysql(uuid, htmlSrc);
				if (!bUpdate) {
					getLogger().error("custom visual updateHtmlToMysql failed, UUID=" + uuid);
				}
			}else{
				getLogger().info("onStopped getUICache(uuid) failed, UUID="+uuid);
			}
		}else{
			getLogger().error("onStopped context.getExtendProperty(\"cvis_uuid\") failed");
		}
	}

	@OnRemoved
	public void OnRemoved(final ProcessContext context) {
		String uuid = context.getExtendProperty("cvis_uuid");
		if (uuid != null && !uuid.isEmpty()) {
			// 从数据库删除缓存的数据
			boolean bDel = deleteHtmlToMysql(uuid);
			if (!bDel) {
				getLogger().error("custom visual deleteHtmlToMysql failed, UUID=" + uuid);
			}
		}else{
			getLogger().error("OnRemoved context.getExtendProperty(\"cvis_uuid\") failed");
		}
	}

	@Override
	public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
		if (session != null) {
			FlowFile flowFile = session.get();
			if (flowFile != null) {
				final byte[] content = new byte[(int) flowFile.getSize()];
				session.read(flowFile, new InputStreamCallback() {

					public void process(final InputStream in) throws IOException {
						StreamUtils.fillBuffer(in, content, true);
					}
				});
				String inputJson = "";
				try {
					inputJson = new String(content, "UTF-8");
				} catch (UnsupportedEncodingException e) {
					e.printStackTrace();
				}
				try {

					Map<String, String> stateMap = Maps
							.newHashMap(context.getStateManager().getState(Scope.LOCAL).toMap());
					stateMap.put("cur_data", inputJson);
					context.getStateManager().replace(context.getStateManager().getState(Scope.LOCAL), stateMap,
							Scope.LOCAL);
				} catch (IOException e) {
					e.printStackTrace();
				}

				session.remove(flowFile);
			}
		}
	}

	public static FlowFile writeToSession(final String resultJson, ProcessSession session, FlowFile flowFile) {
		return session.write(flowFile, new OutputStreamCallback() {
			public void process(OutputStream out) throws IOException {
				out.write(resultJson.getBytes("UTF-8"));
			}
		});
	}

	public static String getWebIP() {
		List<String> listIP = new ArrayList<String>();
		Enumeration allNetInterfaces = null;
		try {
			allNetInterfaces = NetworkInterface.getNetworkInterfaces();
		} catch (SocketException e1) {
			e1.printStackTrace();
		}
		InetAddress ip = null;
		while (allNetInterfaces.hasMoreElements()) {
			NetworkInterface netInterface = (NetworkInterface) allNetInterfaces.nextElement();
			Enumeration addresses = netInterface.getInetAddresses();
			while (addresses.hasMoreElements()) {
				ip = (InetAddress) addresses.nextElement();
				if (ip != null && ip instanceof Inet4Address) {
					listIP.add(ip.getHostAddress());
				}
			}
		}
		for (String oneIP : listIP) {
			if (oneIP.startsWith("10.") || oneIP.startsWith("172.") || oneIP.startsWith("192.168.")
					|| oneIP.equals("127.0.0.1")) {
				//
			} else {
				return oneIP;
			}
		}

		return "127.0.0.1";
	}
/*
	private boolean setUICache(String uuid, String htmlJsSrc) {
		Map<String, String> headers = new HashMap<>();
		headers.put("posttype", "addcache");
		headers.put("uuid", uuid);

		String retPost = HttpTools.PostHttp("http://localhost:8080/nifi-customvisual-ui-1.0-SNAPSHOT/custom_visual",
				headers, htmlJsSrc);

		return (retPost != null && retPost.equals("ok"));
	}
*/
	private String getUICache(String uuid) {
		String htmlJsSrc = null;

		Map<String, String> headers = new HashMap<>();
		headers.put("posttype", "getcache");
		headers.put("uuid", uuid);

		String retPost = HttpTools.PostHttp("http://localhost:8080/nifi-hlcustomerUI-ui-1.0-SNAPSHOT/customvisualui/custom_visual",
				headers, "");
		if (retPost != null && !retPost.startsWith("error")) {
			
			//需要解码，因为原始内容被html编码了
			htmlJsSrc = retPost;
		}

		return htmlJsSrc;
	}

	// 添加html cache到mysql中
	public static boolean addHtmlToMysql(String uuid, String project_id, String user_id, String studio_id,
			String htmlsrc) {
		Connection conn = ConnectDB();
		PreparedStatement st = null;
		int iUpdate = -1;
		try {
			st = (PreparedStatement) conn.prepareStatement(
					"INSERT INTO " + "uuid2html(uuid,project_id,user_id,studio_id,htmlsrc) " + "VALUES (?,?,?,?,?)");

			st.setString(1, uuid);
			st.setString(2, project_id);
			st.setString(3, user_id); // 客户名+主题名+信源为联合主键

			st.setString(4, studio_id);
			st.setString(5, htmlsrc);

			iUpdate = st.executeUpdate();
		} catch (SQLException ex) {
			ex.printStackTrace();

		} finally {
			try {
				if (st != null) {
					st.close();
				}
				conn.close();

			} catch (SQLException ex) {
				ex.printStackTrace();
			}
		}
		return iUpdate >= 0;
	}

	// 从数据库中加载 uuid对应的html
	public static String getHtmlFromMysql(String uuid) {
		String htmlSrc = null;
		Connection conn = ConnectDB();
		try {
			String sql = "select htmlsrc from uuid2html WHERE uuid=\"" + uuid + "\";";
			Statement stmt = (Statement) conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY,
					ResultSet.HOLD_CURSORS_OVER_COMMIT);
			stmt.setFetchSize(Integer.MIN_VALUE);
			ResultSet result = stmt.executeQuery(sql);
			if (result.next()) {
				htmlSrc = result.getString("htmlsrc");
			}
			result.close();

			conn.close();
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			try {
				conn.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		return htmlSrc;
	}

	// 更新数据库中 uuid对应的html
	public static boolean updateHtmlToMysql(String uuid, String htmlSrc) {
		Connection conn = ConnectDB();
		PreparedStatement st = null;
		int iUpdate = -1;
		try {
			st = (PreparedStatement) conn.prepareStatement("UPDATE " + "uuid2html SET htmlsrc=? WHERE uuid=?;");

			st.setString(1, htmlSrc);
			st.setString(2, uuid);

			iUpdate = st.executeUpdate();
		} catch (SQLException ex) {
			ex.printStackTrace();

		} finally {
			try {
				if (st != null) {
					st.close();
				}
				conn.close();

			} catch (SQLException ex) {
				ex.printStackTrace();
			}
		}
		return iUpdate >= 0;
	}

	// 删除数据库中 uuid对应的html
	public static boolean deleteHtmlToMysql(String uuid) {
		Connection conn = ConnectDB();
		PreparedStatement st = null;
		int iUpdate = -1;
		try {
			st = (PreparedStatement) conn.prepareStatement("UPDATE " + "uuid2html SET isdel=1 WHERE uuid=?;");

			st.setString(1, uuid);

			iUpdate = st.executeUpdate();
		} catch (SQLException ex) {
			ex.printStackTrace();

		} finally {
			try {
				if (st != null) {
					st.close();
				}
				conn.close();

			} catch (SQLException ex) {
				ex.printStackTrace();
			}
		}
		return iUpdate >= 0;
	}

	/*
	 * # mysql配置：连接库、用户名、密码
	 * mysqlUrl=jdbc:mysql://192.168.252.43:3308/Online?characterEncoding
	 * =utf8&selectMethod=cursor mysqlUser=root mysqlPwd=1234
	 */
	/*
	 * 连接数据库
	 */
	public static Connection ConnectDB() {
		String strServerDB = "jdbc:mysql://118.89.215.212:3306/customvisual?characterEncoding=utf8&selectMethod=cursor";
		String strUser = "feitu";
		String strPwd = "hylanda";

		Connection connect = null;
		try {
			String CLASSFORNAME = "com.mysql.jdbc.Driver";
			Class.forName(CLASSFORNAME);
			connect = DriverManager.getConnection(strServerDB, strUser, strPwd);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return connect;
	}
}
